use crate::socket_capnp::{map, peer_http_parts, peer_http_request};
use async_stream::stream;
use async_trait::async_trait;

use capnp::message::TypedBuilder;
use futures::{SinkExt, StreamExt};
use rsocket_rust::{
    prelude::{Flux, Payload, RSocket, RSocketFactory},
    runtime, Result,
};
use rsocket_rust_transport_tcp::TcpClientTransport;

use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use std::vec;
use tokio::sync::mpsc;
use tokio::{sync::Mutex, time::timeout};
use tracing_subscriber::fmt::format;
pub const DEFAULT_MIME_TYPE: &str = "application/binary";

pub use rrai_desktop_sdk_common::devices::{
    devices_device_get_peer_id, devices_device_login, devices_device_search,
};

pub async fn devices_device_request(
    request: tauri::http::Request<Vec<u8>>,
) -> anyhow::Result<Vec<u8>> {
    let mut url = request.uri().clone();
    let method = request.method();
    let version = request.version();
    let body = request.body();
    let request_headers = request.headers();

    //
    tracing::debug!("{:#?}|{:#?}", url, request);

    let host = url.host().map_or("localhost", |v| v);
    let port = url.port_u16().map_or(80, |v| v);
    let path = url.path();
    let query = url.query().map_or(String::new(), |v| format!("?{}", v));

    //host:  peer_id.host
    let (peer_id, host) =
        host.split_at(host.find(".").map_or(Err(anyhow::anyhow!("")), |v| Ok(v))?);

    let (_, host) = host.split_at(1);
    //
    tracing::debug!("{},[{}],{}", peer_id, host, port);

    //
    let src_origin = format!("http://");
    let new_origin = format!("");

    let mut message = ::capnp::message::Builder::new_default();

    let mut request_builder = TypedBuilder::<peer_http_request::Owned>::new_default();

    let mut root = request_builder.get_root()?;
    // 设置节点ID
    root.set_peer_id(capnp::text::Reader(peer_id.as_bytes()));

    // headers
    let mut parts_builder = TypedBuilder::<peer_http_parts::Owned>::new_default();
    let mut parts_root = parts_builder.get_root()?;

    //URI
    parts_root.set_uri(capnp::text::Reader(
        format!("http://{}:{}{}{}", host, port, path, query).as_bytes(),
    ));

    //Method
    parts_root.set_method(capnp::text::Reader(
        method.as_str().to_uppercase().as_bytes(),
    ));
    //Version
    parts_root.set_version(capnp::text::Reader(
        format!("{:?}", version).to_uppercase().as_bytes(),
    ));

    //headers
    let mut headers_map_builder =
        TypedBuilder::<map::Owned<capnp::text::Owned, capnp::text::Owned>>::new_default();

    let headers = message.init_root::<map::Builder<capnp::text::Owned, capnp::text::Owned>>();
    let mut entries = headers.init_entries(request_headers.len().try_into()?);
    let mut temp = 0;
    request_headers
        .iter()
        .for_each(|(header_key, header_value)| {
            let mut entry = entries.reborrow().get(temp);
            //referer
            if header_key.as_str().to_lowercase() == "referer" {
                // entry.set_key(capnp::text::Reader(b"referer"));
                // entry.set_value(capnp::text::Reader(value.as_bytes()));
            }
            if header_key.as_str().to_lowercase() == "host" {
                entry.set_key(capnp::text::Reader(b"referer"));

                entry.set_value(capnp::text::Reader(
                    format!("{}:{}", host, port).as_str().as_bytes(),
                ));
            } else {
                if let Ok(value) = header_value.to_str() {
                    entry.set_key(capnp::text::Reader(header_key.as_str().as_bytes()));
                    entry.set_value(capnp::text::Reader(value.as_bytes()));
                }
            }
            temp = temp + 1;
        });

    headers_map_builder
        .get_root()?
        .set_entries(entries.into_reader())?;
    parts_root.set_headers(headers_map_builder.get_root_as_reader()?)?;

    //extensions
    let extensions_map_builder =
        TypedBuilder::<map::Owned<capnp::text::Owned, capnp::text::Owned>>::new_default();

    parts_root.set_extensions(extensions_map_builder.get_root_as_reader()?)?;
    //head
    root.set_head(parts_builder.get_root_as_reader()?)?;

    //routers
    let routers_builder = TypedBuilder::<capnp::text_list::Owned>::new_default();
    root.set_routers(routers_builder.get_root_as_reader()?)?;

    //body
    root.set_body(body);

    let mut buffer = vec![];
    capnp::serialize_packed::write_message(&mut buffer, request_builder.borrow_inner())?;

    tracing::debug!("数据封装完成:{}", buffer.len(),);
    //发送请求
    let has = {
        let cache = rrai_desktop_sdk_common::devices::SOCKET_HANDLE.read().await;
        cache.is_some()
    };

    if has {
        tracing::debug!("连接存在,开始获取连接");
        let arc_client = {
            rrai_desktop_sdk_common::devices::SOCKET_HANDLE
                .read()
                .await
                .as_ref()
                .map_or(Err(anyhow::anyhow!("")), |socket| Ok(socket))?
                .clone()
        };
        let client = arc_client.lock().await;
        //发送数据
        let req = Payload::builder()
            .set_metadata(rsocket_meta_data(
                crate::socket_capnp::request_meta_data::Action::Socket,
            )?)
            .set_data(buffer)
            .build();
        tracing::debug!("开始请求:.....");
        if let Ok(response_result) =
            timeout(Duration::from_millis(30000), client.request_response(req)).await
        {
            let response = response_result.map_err(|err| anyhow::anyhow!("{}", err))?;
            tracing::debug!("返回成功");
            match response {
                Some(res_data) => {
                    //TODO metadata
                    let _metadata = res_data.metadata_utf8();

                    let d = res_data.data();
                    match d {
                        Some(body) => Ok(body.to_vec()),
                        None => Ok(vec![]),
                    }
                }
                None => Ok(vec![]),
            }
        } else {
            tracing::debug!("请求超时!");
            //重新建立连接
            Ok(vec![])
        }
    } else {
        tracing::debug!("没有获取到连接");
        Ok(vec![])
    }
}

fn rsocket_meta_data(action: crate::socket_capnp::request_meta_data::Action) -> Result<Vec<u8>> {
    let mut request_meta_data =
        TypedBuilder::<crate::socket_capnp::request_meta_data::Owned>::new_default();
    let mut root = request_meta_data.get_root()?;
    root.set_action(action);

    let mut buffer = vec![];
    capnp::serialize_packed::write_message(&mut buffer, request_meta_data.borrow_inner())?;

    Ok(buffer)
}
